//! Zenop2p 流与 zenoh Link 绑定：实现数据收发映射
use async_trait::async_trait;
use futures::{AsyncReadExt, AsyncWriteExt, StreamExt};
use libp2p::{
    core::ConnectedPoint,
    identity::PeerId,
    swarm::Swarm,
};
use std::sync::Arc;
use std::str::FromStr;
use tokio::sync::Mutex;
use zenoh_link_commons::{LinkAuthId, LinkUnicastTrait};
use zenoh_protocol::core::Locator;
use zenoh_protocol::transport::BatchSize;
use zenoh_result::ZResult;
use crate::zenop2p::zenop2p_behaviour::Zenop2pBehaviour;

/// 简单的链接统计结构体
#[derive(Default, Debug)]
pub struct LinkStats {
    pub rx_bytes: std::sync::atomic::AtomicU64,
    pub tx_bytes: std::sync::atomic::AtomicU64,
}

/// 封装 libp2p 流的 zenoh Link 实现
#[derive(Debug)]
pub struct Zenop2pLink {
    stream: Arc<tokio::sync::Mutex<libp2p::Stream>>,  // 线程安全的 libp2p 流包装
    peer_id: PeerId,                                // 目标节点 ID
    stats: LinkStats,                               // 传输统计
    connected_point: ConnectedPoint,                // 连接点信息（本地/远程地址）
    src_locator: Locator,                           // 源地址Locator
    dst_locator: Locator,                           // 目标地址Locator
}

impl Zenop2pLink {
    /// 创建新的 Zenop2pLink
    pub fn new(stream: libp2p::Stream, peer_id: PeerId, connected_point: ConnectedPoint) -> Self {
        // 创建源和目标Locator
        let src_addr = match &connected_point {
            ConnectedPoint::Dialer { address, .. } => address.to_string(),
            ConnectedPoint::Listener { ref local_addr, .. } => local_addr.to_string(),
        };
        let dst_addr = format!("{}:{}", peer_id, connected_point.get_remote_address());
        
        let src_locator = Locator::new("zenop2p", src_addr, "").unwrap();
        let dst_locator = Locator::new("zenop2p", dst_addr, "").unwrap();
        
        Self {
            stream: Arc::new(tokio::sync::Mutex::new(stream)),
            peer_id,
            stats: LinkStats::default(),
            connected_point,
            src_locator,
            dst_locator,
        }
    }

    /// 读取长度前缀数据（内部实现）
    async fn read_length_prefixed_internal(&self) -> Result<Vec<u8>, std::io::Error> {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        
        let mut stream = self.stream.lock().await;
        
        // 读取长度前缀（u16）
        let mut len_bytes = [0u8; 2];
        stream.read_exact(&mut len_bytes).await?;
        let len = u16::from_be_bytes(len_bytes) as usize;
        
        // 读取数据
        let mut data = vec![0u8; len];
        stream.read_exact(&mut data).await?;
        
        Ok(data)
    }

    /// 写入长度前缀数据（内部实现）
    async fn write_length_prefixed_internal(&self, buffer: &[u8]) -> Result<(), std::io::Error> {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        
        let mut stream = self.stream.lock().await;
        
        // 写入长度前缀（u16）
        let len = buffer.len() as u16;
        stream.write_all(&len.to_be_bytes()).await?;
        
        // 写入数据
        stream.write_all(buffer).await?;
        
        Ok(())
    }
}

/// 实现 zenoh LinkUnicastTrait 接口：核心数据收发映射
#[async_trait]
impl LinkUnicastTrait for Zenop2pLink {
    /// 从 libp2p 流读取数据（映射到 zenoh read）
    async fn read(&self, buffer: &mut [u8]) -> ZResult<usize> {
        // 读取长度前缀数据（解决粘包）
        let data = self.read_length_prefixed_internal().await.map_err(|e| {
            tracing::error!("[Zenop2pLink] 读取失败: {}", e);
            Box::new(zenoh_result::zerror!("读取失败: {}", e)) as zenoh_result::Error
        })?;

        let n = data.len().min(buffer.len());
        buffer[..n].copy_from_slice(&data[..n]);
        self.stats.rx_bytes.fetch_add(n as u64, std::sync::atomic::Ordering::Relaxed);
        Ok(n)
    }

    /// 向 libp2p 流写入数据（映射到 zenoh write）
    async fn write(&self, buffer: &[u8]) -> ZResult<usize> {
        // 写入长度前缀数据（与读取对应）
        self.write_length_prefixed_internal(buffer).await.map_err(|e| {
            tracing::error!("[Zenop2pLink] 写入失败: {}", e);
            Box::new(zenoh_result::zerror!("写入失败: {}", e)) as zenoh_result::Error
        })?;

        self.stats.tx_bytes.fetch_add(buffer.len() as u64, std::sync::atomic::Ordering::Relaxed);
        Ok(buffer.len())
    }

    /// 关闭流
    async fn close(&self) -> ZResult<()> {
        let mut stream = self.stream.lock().await;
        stream.close().await.map_err(|e| Box::new(zenoh_result::zerror!("关闭失败: {}", e)) as zenoh_result::Error)
    }

    // 元数据与统计实现
    fn get_mtu(&self) -> BatchSize {
        1500  // 假设 MTU 为 1500 字节
    }

    fn get_src(&self) -> &Locator {
        &self.src_locator
    }

    fn get_dst(&self) -> &Locator {
        &self.dst_locator
    }

    fn is_reliable(&self) -> bool {
        true  // libp2p 流是可靠的
    }

    fn is_streamed(&self) -> bool {
        true  // libp2p 流是流式的
    }

    fn get_interface_names(&self) -> Vec<String> {
        vec![]  // 暂时返回空列表，需要根据实际网络接口实现
    }

    fn get_auth_id(&self) -> &LinkAuthId {
        static AUTH_ID: LinkAuthId = LinkAuthId::Tcp;  // 使用Tcp作为默认认证ID
        &AUTH_ID
    }

    async fn write_all(&self, buffer: &[u8]) -> ZResult<()> {
        self.write(buffer).await?;
        Ok(())
    }

    async fn read_exact(&self, buffer: &mut [u8]) -> ZResult<()> {
        let mut offset = 0;
        while offset < buffer.len() {
            let n = self.read(&mut buffer[offset..]).await?;
            if n == 0 {
                return Err(Box::new(zenoh_result::zerror!("连接已关闭")) as zenoh_result::Error);
            }
            offset += n;
        }
        Ok(())
    }
}

/// Zenop2p 传输适配器：管理流与 Link 的绑定
#[derive(Clone)]
pub struct Zenop2pTransport {
    swarm: Arc<Mutex<Swarm<Zenop2pBehaviour>>>,  // 共享的 Zenop2p 网络
    local_peer_id: PeerId,                       // 本地节点 ID
}

impl Zenop2pTransport {
    pub fn new(swarm: Arc<Mutex<Swarm<Zenop2pBehaviour>>>, local_peer_id: PeerId) -> Self {
        Self { swarm, local_peer_id }
    }

    /// 打开与目标节点的流，并封装为 Zenop2pLink
    pub async fn open_link(&self, peer_id: PeerId) -> Result<Zenop2pLink, String> {
        let mut swarm = self.swarm.lock().await;
        // 发起连接请求
        swarm.dial(peer_id.clone()).map_err(|e| format!("连接建立失败: {}", e))?;
        
        // 等待连接建立事件
        loop {
            match swarm.select_next_some().await {
                libp2p::swarm::SwarmEvent::ConnectionEstablished { peer_id: connected_peer_id, endpoint, .. } 
                    if connected_peer_id == peer_id => {
                    // 这里需要实现流升级逻辑，暂时返回一个占位符
                    return Err("流升级逻辑需要实现".to_string());
                }
                libp2p::swarm::SwarmEvent::OutgoingConnectionError { peer_id: error_peer_id, error, .. } 
                    if error_peer_id == Some(peer_id) => {
                    return Err(format!("连接错误: {}", error));
                }
                _ => continue,
            }
        }
    }
}